package com.hong.spring.listener;

import lombok.extern.log4j.Log4j2;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.support.ProducerListener;
/**
 *
 * 功能描述: kafka生产监听
 *
 * @param:
 * @return:
 * @auther: csh
 * @date: 2021/3/29 14:47
 */
@Log4j2
public class KafkaProducerListener implements ProducerListener {
    /**
     *
     * 功能描述: 成功响应
     *
     * @param:
     * @return:
     * @auther: csh
     * @date: 2021/3/29 14:46
     */
    public void onSuccess(String topic, Integer partition, Object key, Object value, RecordMetadata recordMetadata) {
        log.info("==========kafka发送数据成功（日志开始）==========");
        log.info("----------topic:"+topic);
        log.info("----------partition:"+partition);
        log.info("----------key:"+key);
        log.info("----------value:"+value);
        log.info("----------RecordMetadata:"+recordMetadata);
        log.info("~~~~~~~~~~kafka发送数据成功（日志结束）~~~~~~~~~~");
    }

    /**
     * 发送消息错误后调用
     */
    public void onError(String topic, Integer partition, Object key, Object value, Exception exception) {
        log.info("==========kafka发送数据错误（日志开始）==========");
        log.info("----------topic:"+topic);
        log.info("----------partition:"+partition);
        log.info("----------key:"+key);
        log.info("----------value:"+value);
        log.info("----------Exception:"+exception);
        log.info("~~~~~~~~~~kafka发送数据错误（日志结束）~~~~~~~~~~");
        exception.printStackTrace();

    }

    /**
     * 方法返回值代表是否启动kafkaProducer监听器
     */
    public boolean isInterestedInSuccess() {
        log.info("kafkaProducer监听器启动");
        return true;
    }
}